import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.TransactionStatisticsBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class DwsTransctionStatisticsWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTransctionStatisticsWindow().start(
                12593,
                4,
                "dws_transaction_transaction_statistics_window",
                Constant.DWD_TRANS_PAY
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String jsonStr) throws Exception {
                        return JSON.parseObject(jsonStr);
                    }
                }
        );

        //以唯一主键分组去重
        KeyedStream<JSONObject, String> idKeyedDs = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));

        //定时器去重 只获取最后一个
        SingleOutputStreamOperator<JSONObject> distinctDs = idKeyedDs.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<JSONObject> jsonObjectValueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ValueStateDescriptor<JSONObject> stateProperties =
                                new ValueStateDescriptor<JSONObject>("jsonObjectValueState", JSONObject.class);
                        jsonObjectValueState = getRuntimeContext().getState(stateProperties);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

                        JSONObject lastJsonObj = jsonObjectValueState.value();
                        if (lastJsonObj == null) {
                            jsonObjectValueState.update(jsonObj);

                            TimerService timerService = ctx.timerService();
                            long currentProcessingTime = timerService.currentProcessingTime();
                            timerService.registerProcessingTimeTimer(currentProcessingTime + 5000L);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObj = jsonObjectValueState.value();
                        out.collect(jsonObj);
                    }
                }
        );

        KeyedStream<JSONObject, String> orderIdKeyedDs =
                distinctDs.keyBy(jsonObj -> jsonObj.getString("order_id"));

        SingleOutputStreamOperator<JSONObject> processDs = orderIdKeyedDs.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<String> lastOrderIdState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor =
                                new ValueStateDescriptor<String>("lastOrderIdState", String.class);
                        lastOrderIdState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObject = new JSONObject();
                        Long ts = jsonObj.getLong("ts") * 1000;
                        String amount = jsonObj.getString("final_amount");
                        String curOrderId = jsonObj.getString("order_id");
                        //上一个orderid，因为以orderId分组，如果第二次orderid等于上一次orderid说明是同一个订单
                        String lastOrderId = lastOrderIdState.value();

                        jsonObject.put("amount", amount);
                        jsonObject.put("orderUser", 1);
                        jsonObject.put("orderCount", 1);
                        jsonObject.put("ts", ts);

                        if (lastOrderId != null && lastOrderId.equals(curOrderId)) {
                            jsonObject.put("orderUser", 0);
                            jsonObject.put("orderCount", 0);
                        }


                        out.collect(jsonObject);

                    }
                }
        );

        //processDs.print();
        SingleOutputStreamOperator<TransactionStatisticsBean> beanDs = processDs.map(
                new MapFunction<JSONObject, TransactionStatisticsBean>() {
                    @Override
                    public TransactionStatisticsBean map(JSONObject value) throws Exception {
                        TransactionStatisticsBean bean = TransactionStatisticsBean.builder()
                                .stt("")
                                .edt("")
                                .curDate("")
                                .orderUser(value.getLong("orderUser"))
                                .orderCount(value.getLong("orderCount"))
                                .amount(value.getDouble("amount"))
                                .ts(value.getLong("ts"))
                                .build();
                        return bean;
                    }
                }
        );

        //beanDs.print();
        SingleOutputStreamOperator<TransactionStatisticsBean> withWatermarkDs = beanDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TransactionStatisticsBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<TransactionStatisticsBean>() {
                                    @Override
                                    public long extractTimestamp(TransactionStatisticsBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
        );

        AllWindowedStream<TransactionStatisticsBean, TimeWindow> windowDs = withWatermarkDs.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<TransactionStatisticsBean> reduceDs = windowDs.reduce(
                new ReduceFunction<TransactionStatisticsBean>() {
                    @Override
                    public TransactionStatisticsBean reduce(TransactionStatisticsBean value1, TransactionStatisticsBean value2) throws Exception {
                        value1.setOrderUser(value1.getOrderUser() + value2.getOrderUser());
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        value1.setAmount(value1.getAmount() + value2.getAmount());
                        return value1;
                    }
                },
                new AllWindowFunction<TransactionStatisticsBean, TransactionStatisticsBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<TransactionStatisticsBean> values, Collector<TransactionStatisticsBean> out) throws Exception {
                        TransactionStatisticsBean statisticsBean = values.iterator().next();
                        TransactionStatisticsBean bean = TransactionStatisticsBean.builder()
                                .stt(DateFormatUtil.tsToDateTime(window.getStart()))
                                .edt(DateFormatUtil.tsToDateTime(window.getEnd()))
                                .curDate(DateFormatUtil.tsToDate(window.getStart()))
                                .orderUser(statisticsBean.getOrderUser())
                                .orderCount(statisticsBean.getOrderCount())
                                .amount(statisticsBean.getAmount())
                                .ts(statisticsBean.getTs())
                                .build();
                        out.collect(bean);
                    }
                }
        );

        //reduceDs.print();
        //3> TransactionStatisticsBean(stt=2024-11-16 13:22:40, edt=2024-11-16 13:22:50, curDate=2024-11-16, orderUser=4, orderCount=4, amount=800.0, ts=1731734569000)
        //2> TransactionStatisticsBean(stt=2024-11-16 13:22:20, edt=2024-11-16 13:22:30, curDate=2024-11-16, orderUser=22, orderCount=22, amount=4400.0, ts=1731734546000)

        SingleOutputStreamOperator<String> Json = reduceDs.map(new BeanToJsonStrMapFunction<>());

        Json.print();
        //{"stt":"2024-11-16 13:26:20","edt":"2024-11-16 13:26:30","cur_date":"2024-11-16","order_user":33,"order_count":33,"amount":6600.0}
        Json.sinkTo(FlinkSinkUtil.getDorisSink("dws_transaction_transaction_statistics_window"));

    }
}
